package org.zjt.register.service.impl;

import com.alibaba.fastjson.JSONObject;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zjt.config.RegisterConfig;
import org.zjt.config.ZKConnectCenter;
import org.zjt.register.service.IRegisterService;
import org.zjt.domain.RegisteNodeData;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by Administrator on 2017/4/13.
 */
@SuppressWarnings("all")
//@Service
public class RegisterServiceImpl implements IRegisterService {

    private static final Logger LOGGER = LoggerFactory.getLogger(RegisterServiceImpl.class);

    static ExecutorService executorService = Executors.newSingleThreadExecutor();

    String ZK_SEPARATOR = "/";

    @Override
    public void loadRouteInfo(ZKConnectCenter zkConnectCenter, RegisterConfig applicationConfig, String RPC_PARENT_PATH) throws Exception{
        CuratorFramework cf = zkConnectCenter.getZKConnect();
        List<String> list = cf.getChildren().forPath(RPC_PARENT_PATH);
        if (list == null) return;
        for (String path : list){
            Set<RegisteNodeData> nodeData;
            if (!RegisterConfig.REGISTER_ROUTE_MAP.containsKey(path))
                RegisterConfig.REGISTER_ROUTE_MAP.put(path,new HashSet<RegisteNodeData>());
            nodeData = RegisterConfig.REGISTER_ROUTE_MAP.get(path);
            List<String> hosts =cf.getChildren().forPath(RPC_PARENT_PATH+ZK_SEPARATOR+path);
            if (hosts == null) continue;
            for (String host : hosts){
                String content = new String(cf.getData().forPath(RPC_PARENT_PATH+ZK_SEPARATOR+path+ZK_SEPARATOR+host));
                RegisteNodeData registeNodeData = JSONObject.parseObject(content , RegisteNodeData.class);
                nodeData.add(registeNodeData);
            }
        }
    }

    public void updateRouteInfo(PathChildrenCacheEvent event,ZKConnectCenter zkConnectCenter ,String RPC_PARENT_PATH ,RegisterConfig applicationConfig) throws Exception{
        switch (event.getType()) {
            case CHILD_ADDED:
                // /rpc/org.zjt.facade.IHelloService/spring-rpc-provider:8080
                LOGGER.info("CHILD_ADDED :" + event.getData().getPath()+"="+new String(event.getData().getData()));
                updateAllChild(zkConnectCenter,event,applicationConfig,RPC_PARENT_PATH);
                break;
            case CHILD_UPDATED:
                LOGGER.info("CHILD_UPDATED :" + event.getData().getPath()+"="+new String(event.getData().getData()));
                //copyonWrite
                updateAllChild(zkConnectCenter,event,applicationConfig,RPC_PARENT_PATH);
                break;
            case CHILD_REMOVED:
                LOGGER.info("CHILD_REMOVED :" + event.getData().getPath()+"="+new String(event.getData().getData()));
                updateAllChild(zkConnectCenter,event,applicationConfig,RPC_PARENT_PATH);
                break;
            default:
                break;
        }
    }


    public void updateAllChild(ZKConnectCenter zkConnectCenter,PathChildrenCacheEvent event ,RegisterConfig applicationConfig ,String RPC_PARENT_PATH)throws Exception{
        CuratorFramework cf = zkConnectCenter.getZKConnect();
        String [] var1 =event.getData().getPath().split("/");
        if (var1.length <3 )
            throw new Exception("zk路径异常");
        String Facede =var1[2];
        Stat stat = zkConnectCenter.getZKConnect().checkExists().forPath(RPC_PARENT_PATH +ZK_SEPARATOR+ Facede);
        if (stat == null){
            RegisterConfig.REGISTER_ROUTE_MAP.remove(Facede);
            stat = zkConnectCenter.getZKConnect().checkExists().forPath(RPC_PARENT_PATH);
            if (stat == null)
                RegisterConfig.REGISTER_ROUTE_MAP.clear();
            return;
        }
        Set<RegisteNodeData> nodeData =new HashSet<>();
        List<String> list = cf.getChildren().forPath(RPC_PARENT_PATH +ZK_SEPARATOR+ Facede);
        for (String path : list){
            String content = new String(cf.getData().forPath(RPC_PARENT_PATH +ZK_SEPARATOR+ Facede+ZK_SEPARATOR+path));
            RegisteNodeData registeNodeData = JSONObject.parseObject(content , RegisteNodeData.class);
            nodeData.add(registeNodeData);
        }
        RegisterConfig.REGISTER_ROUTE_MAP.put(Facede,nodeData);
    }

    @Override
    public void watchRouteInfo(final String RPC_PARENT_PATH,final ZKConnectCenter zkConnectCenter, final RegisterConfig applicationConfig) throws Exception{
        executorService.submit(()->{
            try {
                //4 建立一个PathChildrenCache缓存,第三个参数为是否接受节点数据内容 如果为false则不接受
                PathChildrenCache cache = new PathChildrenCache(zkConnectCenter.getZKConnect(), RPC_PARENT_PATH, true);
                //5 在初始化的时候就进行缓存监听
                cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
                cache.getListenable().addListener((cf , event)-> updateRouteInfo(event, zkConnectCenter , RPC_PARENT_PATH,applicationConfig));

                CuratorFramework cf = zkConnectCenter.getZKConnect();
                PathChildrenCache cacheChild ;
                List<String> list = cf.getChildren().forPath(RPC_PARENT_PATH);
                for (String path : list){
                    cacheChild = new PathChildrenCache(zkConnectCenter.getZKConnect(), RPC_PARENT_PATH+ZK_SEPARATOR+path, true);
                    cacheChild.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
                    cacheChild.getListenable().addListener((cf1 , event)-> updateRouteInfo(event, zkConnectCenter , RPC_PARENT_PATH,applicationConfig));
                }
                Thread.sleep(Integer.MAX_VALUE);
            }catch (Exception e){
                e.printStackTrace();
            }
        });
    }


    @Override
    public void registerRouteInfo(Map<String,RegisteNodeData> data,String serverName , Integer port  ,ZKConnectCenter zkConnectCenter,String RPC_PARENT_PATH)throws Exception{
        if (data.isEmpty())
            return;
        /** data.put(RPCFacade.class.getName(),new RegisteNodeData("localhost",8070, RpcImpl.class.getName())); */
        for (Map.Entry entry : data.entrySet()){
            Stat stat = zkConnectCenter.getZKConnect().checkExists().forPath(RPC_PARENT_PATH+ZK_SEPARATOR+entry.getKey());
            /*if (stat != null)
                zkConnectCenter.getZKConnect().delete().deletingChildrenIfNeeded().forPath(RPC_PARENT_PATH+ZK_SEPARATOR+entry.getKey());*/
            stat = zkConnectCenter.getZKConnect().checkExists().forPath(RPC_PARENT_PATH+ZK_SEPARATOR+entry.getKey()+ZK_SEPARATOR+serverName+":"+port);
            if (stat == null)
                zkConnectCenter.getZKConnect().create().creatingParentContainersIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(
                        RPC_PARENT_PATH+ZK_SEPARATOR+entry.getKey()+ZK_SEPARATOR+serverName+":"+port,
                        JSONObject.toJSONString(entry.getValue()).getBytes());
            else
                zkConnectCenter.getZKConnect().setData().forPath(
                        RPC_PARENT_PATH+ZK_SEPARATOR+entry.getKey()+ZK_SEPARATOR+serverName+":"+port,
                        JSONObject.toJSONString(entry.getValue()).getBytes());

        }
    }


}
